package com.atguigu.gmall.realtime.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

/**
 * Author: Felix
 * Date: 2021/7/30
 * Desc: 对日志数据进行分流操作
 *  启动日志、曝光日志、页面日志
 *  启动日志放到启动侧输出流中
 *  曝光日志放到曝光侧输出流中
 *  页面日志放到主流中
 *  将不同流的数据写回到kafka的dwd主题中
 *
 *
 * 日志数据分流执行流程
 *  -需要启动的进程
 *      zk、kafka、[hdfs]、logger、BaseLogApp
 *  -运行模拟生成日志jar包
 *  -将生成的日志发送给Nginx
 *  -Nginx接收到数据之后，进行请求转发，将请求发送给202、203以及204上的日志采集服务
 *  -日志采集服务对数据进行输出、落盘以及发送到kafka的ods_base_log
 *  -BaseLogApp从ods_base_log读取数据
 *      >结构转换  String->JSONObject
 *      >状态修复  分组、修复
 *      >分流
 *      >将分流后的数据写到kafka的dwd层不同主题中
 */
public class BaseLogApp01 {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);

        //TODO 2.检查点相关设置
        //2.1 开启检查点 (5秒,数据对齐(EXACTLY_ONCE 精准一次性))
        env.enableCheckpointing(5000L,CheckpointingMode.EXACTLY_ONCE);
        //2.2 设置检查点超时时间 (1分钟)
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        //2.3 设置重启策略 (3次 3秒一次)
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,3000L));
        //2.4 设置job取消后，检查点是否保留 (保留) enableExternalizedCheckpoints开启外部持久化存储 (ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION作业取消的时候也会保留外部检查点)
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //2.5 设置状态后端(文件系统)   内存|文件系统|RocksDB
        env.setStateBackend(new FsStateBackend("hdfs://hadoop101:9820/ck/gmall"));
        //2.6 指定操作HDFS的用户 (报错window用户权限的问题)
        System.setProperty("HADOOP_USER_NAME","atguigu");

        //TODO 3.从Kafka中读取数据
        //3.1 声明消费的主题以及消费者组
        String topic = "ods_base_log";
        String groupId = "base_log_app_group";
        //3.2 获取kafka消费者
        FlinkKafkaConsumer<String> kafkaSource = MyKafkaUtil.getKafkaSource(topic, groupId);
        //3.3 读取数据  封装为流
        DataStreamSource<String> kafkaDS = env.addSource(kafkaSource);

        //TODO 4.对读取的数据进行结构的转换   jsonStr->jsonObj
        //匿名内部类方式实现
        SingleOutputStreamOperator<JSONObject> jsonobjDS = kafkaDS.map(
                new MapFunction<String, JSONObject>() {
                    @Override
                    public JSONObject map(String jsonStr) throws Exception {
                        return JSON.parseObject(jsonStr);
                    }
                }
        );
        //lambda表达式实现
        SingleOutputStreamOperator<JSONObject> jsonObjecDS = kafkaDS.map(
                jsonStr -> JSON.parseObject(jsonStr)
        );

        // 方法的默认调用
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.map(JSON::parseObject);
        jsonObjDS.print(">>>");

        //TODO 5.新老访客状态进行修复

        //TODO 6.按照日志类型对日志进行分流

        //TODO 7.将不同流的数据写到kafka的dwd不同的主题中


        env.execute();
    }
}
